package demo7;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;

public class SubjectPartitionJob {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

        // 1. 初始化配置
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","hdfs://hadoop11:8020");
        //2. 创建job
        Job job = Job.getInstance(conf);
        job.setJarByClass(SubjectPartitionJob.class);

        //3. 设置输入格式化工具和输出格式化
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        //4. 设置输入路径和输出路径
        TextInputFormat.addInputPath(job,new Path("/mapreduce/demo10"));
        TextOutputFormat.setOutputPath(job,new Path("/mapreduce/demo10/out"));

        //TextInputFormat.addInputPath(job,new Path(args[0]));
        //TextOutputFormat.setOutputPath(job,new Path(args[1]));

        //5. 设置mapper和reducer
        job.setMapperClass(SubjectPartitionMapper.class);
        job.setReducerClass(SubjectPartitionReducer.class);

        // 6. 设置mapper的kv类型和reducer的kv类型
        job.setMapOutputKeyClass(DescIntWritable.class);
        job.setMapOutputValueClass(Subject.class);

        job.setOutputKeyClass(Subject.class);
        job.setOutputValueClass(DescIntWritable.class);

        //设置分区规则的类：自定义分区
        job.setPartitionerClass(SubjectPartition.class);
        //设置reduce数量( 分区数量  --out)
        job.setNumReduceTasks(4);

        // 7. 启动job
        boolean b = job.waitForCompletion(true);
        System.out.println(b?"成功":"失败");
    }
    //局部计算 输出：
    // key：成绩  value:subject
    //  50      {张三 语文}
    //  30      {李四 语文}
    //  20      {王五 语文}
    static  class SubjectPartitionMapper extends Mapper<LongWritable, Text, DescIntWritable,Subject> {
        @Override
        protected void map(LongWritable key, Text value,Context context) throws IOException, InterruptedException {
            String[] split = value.toString().split("\t");
            context.write(new DescIntWritable(Integer.parseInt(split[2])),new Subject(split[0],split[1]));
        }
    }
  /*  汇总计算  输出：
    张三 语文 50
    李四 语文 30
    王五 语文 20

    张三 数学 50
    李四 数学 30
    王五 数学 20

    张三 英语 50
    李四 英语 30
    王五 英语 20

    张三 数据 50
    李四 数据 30
    王五 数据 20
 */
    static class SubjectPartitionReducer extends Reducer<DescIntWritable,Subject, Subject, DescIntWritable> {
        @Override
        protected void reduce(DescIntWritable key, Iterable<Subject> values, Context context) throws IOException, InterruptedException {
            for (Subject value : values) {
                context.write(value,key);
            }
        }
    }
}









